package com.oracle.violet.first.report

import com.oracle.violet.first.config.ConfigHandler
import com.oracle.violet.first.until.{CaculateUntils, Jpools}
import org.apache.commons.lang.StringUtils
import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

object MediaAnys {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("地域分布")
      .setMaster("local[*]")
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    val sc = new SparkContext(conf)
    val sQLContext = new SQLContext(sc)

    //读取文件
    val dataFrame: DataFrame = sQLContext.read.parquet(ConfigHandler.parquetFilePath)
    //读取app_dict
    val appDict=sc.textFile(ConfigHandler.appDictFilePath)

    val jedis = Jpools.getJedis
    val appDictData= appDict.map(_.split("\t", -1))
      .filter(_.length >= 5)
      .map(fields=>(fields(1),fields(4)))


        val appdictArray = appDictData.collectAsMap()
        // 他不能广播动态的数据
        val apppdictBT = sc.broadcast(appdictArray)

    val sourceCode=dataFrame.map(row => {
      val appid = row.getAs[String]("appid")
      var appname = row.getAs[String]("appname")
      if (StringUtils.isEmpty(appname)) {
        if (StringUtils.isNotEmpty(appid)) {
           //redis

         appname = apppdictBT.value.getOrElse(appid,appid)
        }
      } else {
        appname = "其他"
      }
      (appname, CaculateUntils.sourceData(row))
    })


        sc.stop()
  }

}
